#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Kafka 工具模块
提供 Kafka consumer group 管理等通用功能
"""

import logging
from typing import Dict, Optional
from kafka.admin import KafkaAdminClient, NewPartitions
from kafka.errors import GroupIdNotFoundError, UnknownError, KafkaError

logger = logging.getLogger(__name__)


def delete_consumer_group(kafka_config: Dict, group_id: str) -> bool:
    """
    删除 Kafka consumer group
    
    Args:
        kafka_config: Kafka 配置字典，包含 host, port, username, password 等
        group_id: 要删除的 consumer group ID
        
    Returns:
        bool: 删除成功返回 True，失败返回 False
    """
    admin_client = None
    try:
        # 构建 bootstrap_servers
        extra_config = kafka_config.get('extra_config', {})
        if isinstance(extra_config, dict) and 'bootstrap_servers' in extra_config:
            bootstrap_servers = extra_config['bootstrap_servers']
        else:
            bootstrap_servers = [f"{kafka_config['host']}:{kafka_config['port']}"]
        
        # 构建管理客户端配置
        admin_config = {
            'bootstrap_servers': bootstrap_servers,
            'request_timeout_ms': 30000,
            'api_version': (2, 6, 0)
        }
        
        # 添加认证配置
        protocol = kafka_config.get('protocol') or (extra_config.get('security_protocol') if isinstance(extra_config, dict) else None)
        if protocol:
            admin_config.update({
                'security_protocol': protocol,
                'sasl_mechanism': kafka_config.get('sasl_mechanism', 'PLAIN'),
                'sasl_plain_username': kafka_config.get('username'),
                'sasl_plain_password': kafka_config.get('password')
            })
        
        logger.info(f"🧹 准备删除 consumer group: {group_id}")
        
        # 创建管理客户端
        admin_client = KafkaAdminClient(**admin_config)
        
        # 删除 consumer group
        result = admin_client.delete_consumer_groups([group_id])
        
        # 检查删除结果
        # delete_consumer_groups 返回列表: [(group_id, error_or_none), ...]
        # 如果删除成功，error 为 None；如果失败，error 为错误类
        if isinstance(result, list) and len(result) > 0:
            for deleted_group_id, error in result:
                if deleted_group_id == group_id:
                    if error is None:
                        # 删除成功
                        logger.info(f"✅ Consumer group 删除成功: {group_id}")
                        return True
                    elif error == GroupIdNotFoundError or isinstance(error, type) and issubclass(error, GroupIdNotFoundError):
                        # Group 不存在，也算成功
                        logger.info(f"ℹ️  Consumer group 不存在（可能已被清理）: {group_id}")
                        return True
                    else:
                        # 删除失败
                        logger.warning(f"⚠️  删除 consumer group 失败: {group_id}, 错误: {error}")
                        return False
        
        logger.warning(f"⚠️  删除 consumer group 结果未知: {group_id}")
        return False
            
    except GroupIdNotFoundError:
        # Group 不存在，记录为 info 级别
        logger.info(f"ℹ️  Consumer group 不存在（可能已被清理）: {group_id}")
        return True
        
    except Exception as e:
        # 其他异常，记录错误但不影响主流程
        logger.error(f"❌ 删除 consumer group 时出错: {group_id}, 错误: {str(e)}")
        return False
        
    finally:
        # 确保关闭管理客户端
        if admin_client:
            try:
                admin_client.close()
            except Exception as e:
                logger.warning(f"关闭 Kafka 管理客户端时出错: {e}")


def delete_consumer_group_safe(kafka_config: Dict, group_id: str) -> None:
    """
    安全地删除 consumer group（捕获所有异常，不抛出）
    适用于 finally 块中的清理操作
    
    Args:
        kafka_config: Kafka 配置字典
        group_id: 要删除的 consumer group ID
    """
    try:
        delete_consumer_group(kafka_config, group_id)
    except Exception as e:
        # 完全静默处理，确保不影响主流程
        logger.debug(f"清理 consumer group 时发生异常（已忽略）: {group_id}, {e}")

